/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.channel.ram;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.sabayframework.channel.Lock;
import org.sabayframework.channel.impl.AbstractChannel;

/**
 * Concurrent implementation of Channel.
 * Lock size can't exceed size of the array.
 * 
 * @author Alex Antonau
 * */
public class ConcurrentArrayChannel<T> extends AbstractChannel<T> {

	private final T[] items;
    /** items index for next take, poll or remove */
    private int takeIndex;
    /** items index for next put, offer, or add. */
    private int putIndex;
    /** Number of items in the queue */
    private volatile int count;
    /** closed flag */
    private volatile boolean closed;

    /** Main lock guarding all access */
    private final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    private final Condition notFull;

    /**
     * Creates channel from given array of elements.
     * @SafeVarargs 
     * */
    public ConcurrentArrayChannel(T... items) {
		this(false, items);
	}
    /**
     * Creates channel from given array of elements.
     * @SafeVarargs 
     * @param isBlocking true if this channel should support blocking put operation
     * @param items array of items
     * */
    public ConcurrentArrayChannel(boolean isBlocking, T... items) {
		this(isBlocking, items, items.length);
	}
    private ConcurrentArrayChannel(boolean isBlocking, T[] items, int count) {
		this.items = items;
        this.lock = new ReentrantLock();
        notEmpty = lock.newCondition();
        notFull = isBlocking ? lock.newCondition() : null;
        this.count = count;
    }

    /**
     * Creates empty array channel
     * */
	public ConcurrentArrayChannel(int size) {
		this(size, false);
	}
    /**
     * Creates empty array channel
     * */
	@SuppressWarnings("unchecked")
	public ConcurrentArrayChannel(int size, boolean isBlocking) {
		this(isBlocking, (T[])new Object[size],0);
	}

	private final class ReadLock implements Lock<T>{
    	private int locked;
		ReadLock(final int size) {
			count-=size;
			locked = size;
		}
		@Override
		public int size() {
			return locked;
		}

		@Override
		public T get(final int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (takeIndex + index) % items.length;
			return items[i];
		}
		@Override
		public void set(final int index,final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (takeIndex + index) % items.length;
			items[i] = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			for(int i=0;i<locked;++i){
				res[i] = items[(takeIndex+i)%items.length];
			}
			return res;
		}
		@Override
		public void unlock() {
			if(locked != -1){
				for(int i=0;i<locked;++i)
					items[(takeIndex+i)%items.length] = null;
				takeIndex = (takeIndex+locked) % items.length;
				locked = -1;
				if(notFull != null) notFull.signalAll();
				lock.unlock();
			}
		}
		
		@Override
		public void cancel() {
			if(locked != -1){
				count+=locked;
				locked = -1;
				lock.unlock();
			}
		}
    }
    
	@Override
	public final Lock<T> getReadLock(final int size) {
		if(size > items.length) throw new IllegalArgumentException(String.format("Lock size %d exceeds array size %d", size, items.length));
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		ReadLock result = null;
		if(lock.tryLock()){
			try{
				if(size <= count){
					result = new ReadLock(size);
				}
			} finally {
				if(result == null) lock.unlock();
			}
		}
		return result;
	}

	private final class WriteLock implements Lock<T>{
		private int locked;
		WriteLock(final int size) {
			count+=size;
			locked = size;
		}
		@Override
		public int size() {
			return locked;
		}
		@Override
		public T get(final int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (putIndex + index) % items.length;
			return items[i];
		}
		@Override
		public void set(final int index,final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (putIndex + index) % items.length;
			items[i] = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			for(int i=0;i<locked;++i){
				res[i] = items[(putIndex+i)%items.length];
			}
			return res;
		}

		@Override
		public void unlock() {
			if(locked !=-1){
				putIndex = (putIndex+locked) % items.length;
				locked = -1;
				notEmpty.signalAll();
				lock.unlock();
			}
		}
		
		@Override
		public void cancel() {
			if(locked!=-1){
				for(int i=0;i<locked;++i){
					items[(putIndex+i)%items.length] = null;
				}
				count-=locked;
				locked = -1;
				lock.unlock();
			}
		}
	}
	@Override
	public final Lock<T> getWriteLock(final int size) {
		if(size > items.length) throw new IllegalArgumentException(String.format("Lock size %d exceeds array size %d", size, items.length));
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		WriteLock result = null;
		if(lock.tryLock()){
			try{
				if(size <= freeSize()){
					result = new WriteLock(size);
				}
			} finally {
				if(result == null) lock.unlock();
			}
		}
		return result;
	}

	@Override
	public boolean isEmpty() {
		return count == 0;
	}

	@Override
	public boolean isFull() {
		return count == items.length;
	}

	@Override
	public int size() {
		return count;
	}

	@Override
	public int capacity() {
		return items.length;
	}

	@Override
	public int freeSize() {
		if(closed)return 0;
		return items.length-count;
	}

	@Override
	public void close() {
		if(closed) return;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
       		closed = true;
       		notEmpty.signalAll();
       		if(notFull!=null) notFull.signalAll();
        } finally {
            lock.unlock();
        }
	}

	@Override
	public boolean isClosed() {
		return closed;
	}

	@Override
	public final Lock<T> waitForReadLock(int size, long timeout)	throws InterruptedException {
        final ReentrantLock lock = this.lock;
        long time=System.currentTimeMillis();
        ReadLock readLock = null;
        if(lock.tryLock(timeout, TimeUnit.MILLISECONDS)){
	        try {
            	timeout -= (System.currentTimeMillis()-time);
	            try {
	                while (timeout > 0 && !closed && count < size){
	                	time=System.currentTimeMillis();
	                	notEmpty.await(timeout, TimeUnit.MILLISECONDS);
	                	timeout -= (System.currentTimeMillis()-time);
	                }
	            } catch (InterruptedException ie) {
	            	notEmpty.signal(); // propagate to non-interrupted thread
	                throw ie;
	            }
				if(size <= count && !closed){
					readLock = new ReadLock(size);
					return readLock;
				}
	            
	        } finally {
	            if(readLock == null) lock.unlock();
	        }
        }
        return getReadLock(size);
	}

	/**
	 * Returns elements are currently in the channel.
	 * Not a thread safe method.
	 * */
	@SuppressWarnings("unchecked")
	public T[] toArray() {
		T[] res = (T[])new Object[count];
		for(int i=0;i<count;++i){
			res[i] = items[(takeIndex+i)%items.length];
		}
		return res;
	}

	/**
	 * Puts given item into the channel. Waits while space is available.
	 * @throws IllegalStateException if channel is closed or interrupted
	 * */
	public void put(T elem){
		if(notFull == null) throw new IllegalStateException("Channel was created without blocking call support");
		lock.lock();
		try{
            try {
                while (!closed && items.length == count){
                	notFull.await();
                }
            } catch (InterruptedException ie) {}
			
			if(freeSize() >= 1){
				count++;
				final int i = putIndex;
				items[i] = elem;
				putIndex = (i+1) % items.length;
				notEmpty.signalAll();
			}else{
				throw new IllegalStateException(); 
			}
			
		}finally{
			lock.unlock();
		}
	}
}
